package com.hfzy.ihk.web.callhistoryold.jobs.simple;


import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.hfzy.ihk.common.util.string.StringTools;
import com.hfzy.ihk.web.callhistoryold.biz.CallHistoryOldBiz;
import com.hfzy.ihk.web.callhistoryold.constant.RedisKey;
import com.hfzy.ihk.web.callhistoryold.entity.CallHistoryOld;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Slf4j
@Component("callHisBuildFromHbJob")
public class CallHisBuildFromHbJob implements SimpleJob {

    @Autowired
    CallHistoryOldBiz callHistoryOldBiz;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Value("${job.callHis.build.limit}")
    int limit;

    //记录分片ShardingItem与其对应的userId和startRowKey的关系，用以在hbase拉取数据,需保证每个userId只被1个分读取到
    Map<Integer,IndexData> indexMap = new HashMap();

    @Override
    public void execute(ShardingContext context) {

        //redis中这个参数控制是否跑数，当需要停止跑数的时候只要更新redis中这个参数即可
        String isRun = stringRedisTemplate.opsForValue().get(RedisKey.IS_RUN_BUILD_CALLHIS);
        if(isRun != null && isRun.equals("1")) {

            long startTime = System.currentTimeMillis();
        /*
          为保证多分片并发读取不会重复读取数据，
          需根据分片区分读取的数据，保证一个userId只会被一个分片读取到;
         */
            int item = context.getShardingItem();
            log.info("CallHisBuildFromHbJob:分片:" + item);
            IndexData data = indexMap.get(item);
            try {
                if (data == null) {
                    //从redis中读取以List结构保存的userId列表，利用redis单线程的特性，每次从列表中lpop出一个userId，这样保证一个userId只会被读取1次
                    String userId = stringRedisTemplate.opsForList().leftPop(RedisKey.CALLHIS_UID_LIST);
                    if (StringTools.isNotempty(userId)) {
                        data = new IndexData(Integer.parseInt(userId), null);
                        indexMap.put(item, data);
                    }
                }


                if (data != null) {
                    log.info("CallHisBuildFromHbJob:分片:" + item + ",userId:" + data.getUserId());
                    List<CallHistoryOld> list = callHistoryOldBiz.getDatasFromHBaseByUserId(data.getUserId(), limit, data.getStartRowKey());

                    //循环再处理字段，例如一些空格符等等
                    resetDatas(list);

                    if (list.size() > 0) {
                        callHistoryOldBiz.insertBatch(list, limit);
                        log.info("CallHisBuildFromHbJob:分片:" + item + ",userId:" + data.getUserId() + ",已写入:" + list.size() + "条数据,耗时:" + (System.currentTimeMillis() - startTime));
                    }
                    if (list.size() >= limit) {
                        data.setStartRowKey(list.get(list.size() - 1).getSessionId());
                    } else {
               /* 当list的size小于limit的时候，说明已经从hbase中取到最后一批，
                  因此把indexMap对应分区的数据清掉，
                  以便该分区下一个线程执行的时候可以从redis读取下一个userId*/
                        indexMap.remove(item);
                        log.info("CallHisBuildFromHbJob:分片:" + item + ",userId:" + data.getUserId() + ",已跑完最后一批");
                    }
                }
            }catch (Exception e) {
                if (data != null && data.getUserId() != null) {
                    stringRedisTemplate.opsForSet().add(RedisKey.BUILD_CALLHIS_EX_UIDS,data.getUserId() + "");
                }
                log.error("报错:CallHisBuildFromHbJob,userId:" + ((data != null && data.getUserId() != null)?(data.getUserId() + ""):"无"),e);
                e.printStackTrace();
            }
        }
    }

    private void resetDatas(List<CallHistoryOld> list){
        for(CallHistoryOld callHistoryOld : list){
            callHistoryOld.setCallFromType(callHistoryOld.getCallFromType().trim());
            callHistoryOld.setCallChannel(callHistoryOld.getCallChannel().trim());
        }
    }

    @Data
    private class IndexData{
        public IndexData(Integer userId, String startRowKey) {
            this.userId = userId;
            this.startRowKey = startRowKey;
        }

        private Integer userId;

        private String startRowKey;
    }
}
